package com.lwl.concurrency.forkjoin;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.TimeUnit;

/**
 * 带有返回的值的任务
 * 
 * @author lwl 2018年1月14日 下午11:22:56
 */
public class ForkJoinTaskTest {

	/**
	 * 创建一个Document类，它将产生用来模拟文档的字符串的二维数组。
	 */
	public static class Document {
		private String words[] = { "the", "hello", "goodbye", "packt", "java", "thread", "pool", "random", "class",
				"main" };

		public String[][] generateDocument(int numLines, int numWords, String word) {
			// 首先，创建生成这个文档必需的对象：字符串二维对象和生成随机数的Random对象。
			int counter = 0;
			String document[][] = new String[numLines][numWords];
			Random random = new Random();
			// 用字符串填充这个数组。存储在每个位置的字符串是单词数组的随机位置，统计这个程序将要在生成的数组中查找的单词出现的次数。你可以使用这个值来检查程序是否执行正确。
			for (int i = 0; i < numLines; i++) {
				for (int j = 0; j < numWords; j++) {
					int index = random.nextInt(words.length);
					document[i][j] = words[index];
					if (document[i][j].equals(word)) {
						counter++;
					}
				}
			}
			// 将单词出现的次数写入控制台，并返回生成的二维数组。
			System.out.println("DocumentMock: The word appears " + counter + " times in the document");
			return document;
		}
	}

	/**
	 * 创建一个DocumentTask类，指定它继承RecursiveTask类，并参数化为Integer类型。
	 * 该类将实现统计单词在一组行中出现的次数的任务。
	 */
	public static class DocumentTask extends RecursiveTask<Integer> {
		private static final long serialVersionUID = 1L;
		private String document[][];
		private int start, end;
		private String word;

		public DocumentTask(String document[][], int start, int end, String word) {
			this.document = document;
			this.start = start;
			this.end = end;
			this.word = word;
		}

		@Override
		protected Integer compute() {
			int result = 0;
			if (end - start < 10) {
				result = processLines(document, start, end, word);
				// 否则，用两个对象分解行组，创建两个新的DocumentTask对象用来处理这两个组，并且在池中使用invokeAll()方法来执行它们。
			} else {
				int mid = (start + end) / 2;
				DocumentTask task1 = new DocumentTask(document, start, mid, word);
				DocumentTask task2 = new DocumentTask(document, mid, end, word);
				invokeAll(task1, task2);
				// 然后，使用groupResults()方法将这两个任务返回的结果相加。最后，返回这个任务统计的结果。
				try {
					result = groupResults(task1.get(), task2.get());
				} catch (InterruptedException | ExecutionException e) {
					e.printStackTrace();
				}
			}
			return result;
		}

		private Integer processLines(String[][] document, int start, int end, String word) {
			List<LineTask> tasks = new ArrayList<LineTask>();
			for (int i = start; i < end; i++) {
				LineTask task = new LineTask(document[i], 0, document[i].length, word);
				tasks.add(task);
			}
			invokeAll(tasks);
			int result = 0;
			for (int i = 0; i < tasks.size(); i++) {
				LineTask task = tasks.get(i);
				try {
					result = result + task.get();
				} catch (InterruptedException | ExecutionException e) {
					e.printStackTrace();
				}
			}
			return result;
		}

		private Integer groupResults(Integer number1, Integer number2) {
			Integer result;
			result = number1 + number2;
			return result;
		}

	}

	/**
	 * 创建LineTask类，指定它继承RecursiveTask类，并参数化为Integer类型。这个类将实现统计单词在一行中出现的次数的任务。
	 */
	public static class LineTask extends RecursiveTask<Integer> {
		private static final long serialVersionUID = 1L;
		private String line[];
		private int start, end;
		private String word;

		public LineTask(String line[], int start, int end, String word) {
			this.line = line;
			this.start = start;
			this.end = end;
			this.word = word;
		}

		@Override
		protected Integer compute() {
			Integer result = null;
			if (end - start < 100) {
				result = count(line, start, end, word);
			} else {
				int mid = (start + end) / 2;
				LineTask task1 = new LineTask(line, start, mid, word);
				LineTask task2 = new LineTask(line, mid, end, word);
				invokeAll(task1, task2);
				try {
					result = groupResults(task1.get(), task2.get());
				} catch (InterruptedException | ExecutionException e) {
					e.printStackTrace();
				}
			}
			return result;
		}

		private Integer count(String[] line, int start, int end, String word) {
			int counter;
			counter = 0;
			for (int i = start; i < end; i++) {
				if (line[i].equals(word)) {
					counter++;
				}
			}
			try {
				Thread.sleep(10);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			return counter;
		}

		private Integer groupResults(Integer number1, Integer number2) {
			Integer result;
			result = number1 + number2;
			return result;
		}
	}

	public static void main(String[] args) {
		Document mock = new Document();
		String[][] document = mock.generateDocument(100, 1000, "the");
		DocumentTask task = new DocumentTask(document, 0, 100, "the");
		ForkJoinPool pool = new ForkJoinPool();
		pool.execute(task);
		do {
			System.out.printf("******************************************\n");
			System.out.printf("Main: Parallelism: %d\n", pool.getParallelism());
			System.out.printf("Main: Active Threads: %d\n", pool.getActiveThreadCount());
			System.out.printf("Main: Task Count: %d\n", pool.getQueuedTaskCount());
			System.out.printf("Main: Steal Count: %d\n", pool.getStealCount());
			System.out.printf("******************************************\n");
			try {
				TimeUnit.SECONDS.sleep(1);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		} while (!task.isDone());
		pool.shutdown();
		try {
			System.out.printf("Main: The word appears %d in the document", task.get());
		} catch (InterruptedException | ExecutionException e) {
			e.printStackTrace();
		}
		try {
			System.out.printf("Main: The word appears %d in the document", task.get());
		} catch (InterruptedException | ExecutionException e) {
			e.printStackTrace();
		}

	}

}
